package com.fwmagic.spark.core.transformations

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 按照Key进行聚合，只需要传入一个函数和一个初始值，初始值只在局部聚合时使用，全局聚合不使用
  * 和ReduceByKey类似，局部聚合和全局用的是同一个函数。
  */
object FoldByKeyDemo {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf()
                .setAppName(this.getClass.getSimpleName)
                .setMaster("local[*]")

        val sc: SparkContext = new SparkContext(conf)

        val rdd: RDD[(String, Int)] = sc.parallelize(List(("cat", 2), ("cat", 5), ("mouse", 4),
            ("cat", 12), ("dog", 12), ("mouse", 2)), 2)

        /*
            初始值只在局部聚合时使用，全局聚合不使用
            (dog,112)
            (cat,219)
            (mouse,206)
         */
        val res: RDD[(String, Int)] = rdd.foldByKey(100)(_ + _)

        res.collect().foreach(println)

        sc.stop()

    }

}
